package com.zbkj.client.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.springframework.http.MediaType;

/**
 * SSE会话管理器
 * 用于管理SSE连接和消息，支持服务端和客户端模式
 */
@Slf4j
@Component
public class SSESessionManager {

    // 用户ID -> 该用户的所有SSE连接
    private final Map<Integer, CopyOnWriteArrayList<SSESession>> userSessions = new ConcurrentHashMap<>();
    
    // 连接ID -> SSE会话
    private final Map<String, SSESession> sessionMap = new ConcurrentHashMap<>();
    
    // 会话未接收的消息缓存，会话ID -> 消息列表
    private final Map<String, CopyOnWriteArrayList<SSEMessage>> sessionMessages = new ConcurrentHashMap<>();
    
    // 定时清理过期连接的调度器
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    
    // SSE连接有效期（30分钟）
    private static final long SESSION_TIMEOUT = 30 * 60 * 1000; // 30分钟
    
    public SSESessionManager() {
        // 启动定时清理任务，每5分钟执行一次
        scheduler.scheduleAtFixedRate(this::cleanExpiredSessions, 5, 5, TimeUnit.MINUTES);
    }
    
    /**
     * 注册SSE会话
     * @param userId 用户ID
     * @param emitter SSE发射器
     * @return 会话ID
     */
    public String registerSession(Integer userId, SseEmitter emitter) {
        String sessionId = UUID.randomUUID().toString();
        SSESession session = new SSESession(sessionId, userId, emitter, LocalDateTime.now());
        
        // 添加到会话映射
        sessionMap.put(sessionId, session);
        
        // 添加到用户会话列表
        userSessions.computeIfAbsent(userId, k -> new CopyOnWriteArrayList<>()).add(session);
        
        // 设置超时回调
        emitter.onTimeout(() -> {
            log.info("SSE连接超时，用户ID: {}, 会话ID: {}", userId, sessionId);
            removeSession(userId, sessionId);
        });
        
        // 设置完成回调
        emitter.onCompletion(() -> {
            log.info("SSE连接完成，用户ID: {}, 会话ID: {}", userId, sessionId);
            removeSession(userId, sessionId);
        });
        
        // 设置错误回调
        emitter.onError(e -> {
            log.error("SSE连接错误，用户ID: {}, 会话ID: {}, 错误: {}", userId, sessionId, e.getMessage());
            removeSession(userId, sessionId);
        });
        
        log.info("SSE连接已建立，用户ID: {}, 会话ID: {}", userId, sessionId);
        
        // 发送连接成功事件
        try {
            Map<String, Object> data = new ConcurrentHashMap<>();
            data.put("sessionId", sessionId);
            data.put("userId", userId);
            data.put("timestamp", System.currentTimeMillis());
            emitter.send(SseEmitter.event()
                    .name("connected")
                    .data(data)
                    .reconnectTime(3000));
            
            // 发送未接收的消息
            sendPendingMessages(sessionId);
        } catch (IOException e) {
            log.error("发送连接成功事件失败，用户ID: {}, 会话ID: {}, 错误: {}", userId, sessionId, e.getMessage());
            removeSession(userId, sessionId);
        }
        
        return sessionId;
    }
    
    /**
     * 移除SSE会话
     * @param userId 用户ID
     * @param sessionId 会话ID
     */
    public void removeSession(Integer userId, String sessionId) {
        SSESession session = sessionMap.remove(sessionId);
        if (session != null) {
            try {
                session.getEmitter().complete();
            } catch (Exception e) {
                log.error("关闭SSE连接失败，用户ID: {}, 会话ID: {}, 错误: {}", userId, sessionId, e.getMessage());
            }
        }
        
        CopyOnWriteArrayList<SSESession> sessions = userSessions.get(userId);
        if (sessions != null) {
            sessions.removeIf(s -> s.getSessionId().equals(sessionId));
            if (sessions.isEmpty()) {
                userSessions.remove(userId);
            }
        }
        
        // 清理会话消息缓存
        sessionMessages.remove(sessionId);
        
        log.info("SSE连接已移除，用户ID: {}, 会话ID: {}", userId, sessionId);
    }
    
    /**
     * 刷新SSE会话活跃时间
     * @param sessionId 会话ID
     */
    public void refreshSession(String sessionId) {
        SSESession session = sessionMap.get(sessionId);
        if (session != null) {
            session.setLastActiveTime(LocalDateTime.now());
            log.debug("刷新SSE会话活跃时间，会话ID: {}", sessionId);
        }
    }
    
    /**
     * 向用户发送消息
     * @param userId 用户ID
     * @param eventName 事件名称
     * @param data 消息数据
     */
    public void sendMessageToUser(Integer userId, String eventName, Object data) {
        // 创建消息对象
        SSEMessage message = new SSEMessage(eventName, data, LocalDateTime.now());
        
        // 获取用户的所有会话
        CopyOnWriteArrayList<SSESession> sessions = userSessions.get(userId);
        
        // 如果用户没有活跃会话，记录日志并返回
        if (sessions == null || sessions.isEmpty()) {
            log.info("用户 {} 没有活跃的SSE连接，消息已缓存", userId);
            return;
        }
        
        // 发送消息到用户的所有会话
        boolean allFailed = true;
        for (SSESession session : sessions) {
            try {
                session.getEmitter().send(SseEmitter.event()
                        .name(eventName)
                        .data(data)
                        .reconnectTime(3000));
                session.setLastActiveTime(LocalDateTime.now());
                allFailed = false;
            } catch (IOException e) {
                log.error("向用户发送消息失败，用户ID: {}, 会话ID: {}, 错误: {}", userId, session.getSessionId(), e.getMessage());
                // 标记会话为失效，稍后会被清理
                session.setInvalid(true);
            }
        }
        
        // 如果所有会话都发送失败，缓存消息到所有会话
        if (allFailed) {
            for (SSESession session : sessions) {
                cacheMessageForSession(session.getSessionId(), message);
            }
        }
    }
    
    /**
     * 向特定会话发送消息
     * @param sessionId 会话ID
     * @param eventName 事件名称
     * @param data 消息数据
     * @return 是否发送成功
     */
    public boolean sendMessageToSession(String sessionId, String eventName, Object data) {
        SSESession session = sessionMap.get(sessionId);
        if (session == null) {
            log.warn("会话不存在，会话ID: {}", sessionId);
            return false;
        }
        
        if (session.isInvalid()) {
            log.warn("会话已失效，会话ID: {}", sessionId);
            return false;
        }
        
        try {
            session.getEmitter().send(SseEmitter.event()
                    .name(eventName)
                    .data(data)
                    .reconnectTime(3000));
            session.setLastActiveTime(LocalDateTime.now());
            log.debug("向会话发送消息成功，会话ID: {}, 事件: {}", sessionId, eventName);
            return true;
        } catch (IOException e) {
            log.error("向会话发送消息失败，会话ID: {}, 事件: {}, 错误: {}", sessionId, eventName, e.getMessage());
            session.setInvalid(true);
            return false;
        }
    }
    
    /**
     * 向特定会话发送原始数据流（透传模式）
     * @param sessionId 会话ID
     * @param rawData 原始数据
     * @return 是否发送成功
     */
    public boolean sendRawToSession(String sessionId, String rawData) {
        SSESession session = sessionMap.get(sessionId);
        if (session == null) {
            log.warn("会话不存在，会话ID: {}", sessionId);
            return false;
        }
        
        if (session.isInvalid()) {
            log.warn("会话已失效，会话ID: {}", sessionId);
            return false;
        }
        
        try {
            // 使用SseEmitter的send方法发送原始数据
            session.getEmitter().send(rawData, MediaType.TEXT_EVENT_STREAM);
            session.setLastActiveTime(LocalDateTime.now());
            log.debug("向会话透传原始数据成功，会话ID: {}, 数据长度: {}", sessionId, rawData.length());
            return true;
        } catch (IOException e) {
            log.error("向会话透传原始数据失败，会话ID: {}, 错误: {}", sessionId, e.getMessage());
            session.setInvalid(true);
            return false;
        }
    }

    /**
     * 向用户的所有会话发送原始数据流（透传模式）
     * @param userId 用户ID
     * @param rawData 原始数据
     * @return 是否发送成功
     */
    public boolean sendRawToUser(Integer userId, String rawData) {
        CopyOnWriteArrayList<SSESession> sessions = userSessions.get(userId);
        if (sessions == null || sessions.isEmpty()) {
            log.info("用户 {} 没有活跃的SSE连接", userId);
            return false;
        }
        
        boolean allFailed = true;
        for (SSESession session : sessions) {
            if (sendRawToSession(session.getSessionId(), rawData)) {
                allFailed = false;
            }
        }
        
        return !allFailed;
    }
    
    /**
     * 关闭特定会话
     * @param sessionId 会话ID
     */
    public void closeSession(String sessionId) {
        SSESession session = sessionMap.get(sessionId);
        if (session == null) {
            log.warn("会话不存在，无法关闭，会话ID: {}", sessionId);
            return;
        }
        
        try {
            session.getEmitter().complete();
            log.info("会话已关闭，会话ID: {}", sessionId);
        } catch (Exception e) {
            log.error("关闭会话时发生错误，会话ID: {}, 错误: {}", sessionId, e.getMessage());
        } finally {
            removeSession(session.getUserId(), sessionId);
        }
    }
    
    /**
     * 向所有用户广播消息
     * @param eventName 事件名称
     * @param data 消息数据
     */
    public void broadcastMessage(String eventName, Object data) {
        // 创建消息对象
        SSEMessage message = new SSEMessage(eventName, data, LocalDateTime.now());
        
        // 获取所有用户ID
        Set<Integer> userIds = userSessions.keySet();
        
        if (userIds.isEmpty()) {
            log.info("没有活跃用户，广播消息已取消");
            return;
        }
        
        // 向每个用户发送消息
        for (Integer userId : userIds) {
            sendMessageToUser(userId, eventName, data);
        }
        
        log.info("广播消息已发送给 {} 个用户，事件: {}", userIds.size(), eventName);
    }
    
    /**
     * 缓存会话消息
     * @param sessionId 会话ID
     * @param message 消息
     */
    private void cacheMessageForSession(String sessionId, SSEMessage message) {
        sessionMessages.computeIfAbsent(sessionId, k -> new CopyOnWriteArrayList<>()).add(message);
        log.info("消息已缓存，会话ID: {}, 事件: {}", sessionId, message.getEventName());
    }
    
    /**
     * 发送会话未接收的消息
     * @param sessionId 会话ID
     */
    private void sendPendingMessages(String sessionId) {
        CopyOnWriteArrayList<SSEMessage> messages = sessionMessages.get(sessionId);
        if (messages == null || messages.isEmpty()) {
            return;
        }
        
        SSESession session = sessionMap.get(sessionId);
        if (session == null) {
            return;
        }
        
        // 发送所有未接收的消息
        for (SSEMessage message : messages) {
            try {
                session.getEmitter().send(SseEmitter.event()
                        .name(message.getEventName())
                        .data(message.getData())
                        .reconnectTime(3000));
            } catch (IOException e) {
                log.error("发送未接收消息失败，会话ID: {}, 错误: {}", sessionId, e.getMessage());
                return; // 发送失败，停止发送后续消息
            }
        }
        
        // 清空已发送的消息
        messages.clear();
        log.info("已发送所有未接收消息，会话ID: {}", sessionId);
    }
    
    /**
     * 检查用户是否有活跃的SSE连接
     * @param userId 用户ID
     * @return 是否有活跃连接
     */
    public boolean hasActiveSession(Integer userId) {
        CopyOnWriteArrayList<SSESession> sessions = userSessions.get(userId);
        if (sessions == null || sessions.isEmpty()) {
            return false;
        }
        
        // 清理已失效的连接
        sessions.removeIf(session -> session.isInvalid() || !isEmitterActive(session.getEmitter()));
        
        if (sessions.isEmpty()) {
            userSessions.remove(userId);
            return false;
        }
        
        return true;
    }
    
    /**
     * 获取当前活跃连接数
     * @return 活跃连接数
     */
    public int getActiveSessionCount() {
        return userSessions.values().stream()
                .mapToInt(sessions -> (int) sessions.stream()
                        .filter(session -> !session.isInvalid() && isEmitterActive(session.getEmitter()))
                        .count())
                .sum();
    }
    
    /**
     * 获取用户的所有会话ID
     * @param userId 用户ID
     * @return 会话ID列表
     */
    public CopyOnWriteArrayList<String> getUserSessionIds(Integer userId) {
        CopyOnWriteArrayList<SSESession> sessions = userSessions.get(userId);
        if (sessions == null || sessions.isEmpty()) {
            return new CopyOnWriteArrayList<>();
        }
        
        CopyOnWriteArrayList<String> sessionIds = new CopyOnWriteArrayList<>();
        for (SSESession session : sessions) {
            if (!session.isInvalid() && isEmitterActive(session.getEmitter())) {
                sessionIds.add(session.getSessionId());
            }
        }
        
        return sessionIds;
    }
    
    /**
     * 检查SseEmitter是否活跃（兼容JDK8）
     * @param emitter SSE发射器
     * @return 是否活跃
     */
    private boolean isEmitterActive(SseEmitter emitter) {
        try {
            // 在JDK8中，SseEmitter没有isOpen()方法
            // 我们通过检查emitter是否为null和是否已标记为无效来判断
            // 实际的连接状态检查会在发送消息时进行
            return emitter != null;
        } catch (Exception e) {
            // 发生异常，说明连接已断开
            return false;
        }
    }
    
    /**
     * 定时清理过期的SSE连接和消息
     * 每5分钟执行一次
     */
    @Scheduled(fixedRate = 300000) // 5分钟
    public void cleanExpiredSessions() {
        LocalDateTime now = LocalDateTime.now();
        int expiredCount = 0;
        int invalidCount = 0;
        
        // 清理过期的会话
        for (Map.Entry<String, SSESession> entry : sessionMap.entrySet()) {
            SSESession session = entry.getValue();
            
            // 检查会话是否已失效或超过30分钟未活跃
            if (session.isInvalid() || session.getLastActiveTime().plusMinutes(30).isBefore(now)) {
                try {
                    if (session.isInvalid()) {
                        invalidCount++;
                    } else {
                        expiredCount++;
                    }
                    
                    // 关闭连接并移除会话
                    session.getEmitter().complete();
                    removeSession(session.getUserId(), session.getSessionId());
                } catch (Exception e) {
                    log.error("清理过期会话失败，用户ID: {}, 会话ID: {}, 错误: {}", 
                            session.getUserId(), session.getSessionId(), e.getMessage());
                }
            }
        }
        
        // 清理过期的消息（超过30分钟）
        for (Map.Entry<String, CopyOnWriteArrayList<SSEMessage>> entry : sessionMessages.entrySet()) {
            entry.getValue().removeIf(message -> message.getTimestamp().plusMinutes(30).isBefore(now));
            if (entry.getValue().isEmpty()) {
                sessionMessages.remove(entry.getKey());
            }
        }
        
        if (expiredCount > 0 || invalidCount > 0) {
            log.info("已清理 {} 个过期会话和 {} 个失效会话", expiredCount, invalidCount);
        }
    }
    
    /**
     * SSE会话类
     */
    private static class SSESession {
        private final String sessionId;
        private final Integer userId;
        private final SseEmitter emitter;
        private LocalDateTime lastActiveTime;
        private boolean invalid = false;
        
        public SSESession(String sessionId, Integer userId, SseEmitter emitter, LocalDateTime lastActiveTime) {
            this.sessionId = sessionId;
            this.userId = userId;
            this.emitter = emitter;
            this.lastActiveTime = lastActiveTime;
        }
        
        public String getSessionId() {
            return sessionId;
        }
        
        public Integer getUserId() {
            return userId;
        }
        
        public SseEmitter getEmitter() {
            return emitter;
        }
        
        public LocalDateTime getLastActiveTime() {
            return lastActiveTime;
        }
        
        public void setLastActiveTime(LocalDateTime lastActiveTime) {
            this.lastActiveTime = lastActiveTime;
        }
        
        public boolean isInvalid() {
            return invalid;
        }
        
        public void setInvalid(boolean invalid) {
            this.invalid = invalid;
        }
    }
    
    /**
     * SSE消息类
     */
    private static class SSEMessage {
        private final String eventName;
        private final Object data;
        private final LocalDateTime timestamp;
        
        public SSEMessage(String eventName, Object data, LocalDateTime timestamp) {
            this.eventName = eventName;
            this.data = data;
            this.timestamp = timestamp;
        }
        
        public String getEventName() {
            return eventName;
        }
        
        public Object getData() {
            return data;
        }
        
        public LocalDateTime getTimestamp() {
            return timestamp;
        }
    }
}